package com.dyx.source.reactor.multiThreadMultiReactor;

import com.dyx.source.util.ReadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;

public class WorkerReactor implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(WorkerReactor.class);

    private String name;

    private final Selector selector;

    BlockingDeque<Runnable> QUEUE = new LinkedBlockingDeque<>();

    public WorkerReactor(String name) throws IOException {
        this.name = name;
        this.selector = Selector.open();
        new Thread(this, this.name).start();
    }

    public void bind(SocketChannel socketChannel) throws IOException {
        socketChannel.configureBlocking(false);
        final String threadName = Thread.currentThread().getName();
        QUEUE.add(() -> {
            try {
                LOGGER.info("将socketChannel[{}]注册到Worker[{}]的selector之前...", socketChannel, threadName);
                socketChannel.register(selector, SelectionKey.OP_READ);
                LOGGER.info("将socketChannel[{}]注册到Worker[{}]的selector之后...", socketChannel, threadName);
            } catch (ClosedChannelException e) {
                e.printStackTrace();
            }
        });
        selector.wakeup();
    }

    /**
     * When an object implementing interface <code>Runnable</code> is used
     * to create a thread, starting the thread causes the object's
     * <code>run</code> method to be called in that separately executing
     * thread.
     * <p>
     * The general contract of the method <code>run</code> is that it may
     * take any action whatsoever.
     *
     * @see Thread#run()
     */
    @Override
    public void run() {
        final String threadName = Thread.currentThread().getName();
        try {
            while (true) {
                LOGGER.info("Worker[{}]在select方法之前...", threadName);
                selector.select();
                LOGGER.info("Worker[{}]在select方法之后...", threadName);

                final Runnable runnable = QUEUE.poll();
                if (runnable != null) {
                    runnable.run();
                }

                final Set<SelectionKey> selectionKeys = selector.selectedKeys();
                final Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    final SelectionKey selectionKey = iterator.next();
                    iterator.remove();
                    if (selectionKey.isReadable()) {
                        final SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                        LOGGER.info("selectionKey中的socketChannel[{}]可读...", socketChannel);
                        ReadUtils.readFromClient(socketChannel, selectionKey);
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
